package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/louloulin/dataflare/pkg/processing/data"
	"github.com/louloulin/dataflare/pkg/processing/processor"
	"github.com/louloulin/dataflare/pkg/processing/unit"
)

func main() {
	// 创建临时目录
	tempDir, err := os.MkdirTemp("", "processing-example")
	if err != nil {
		log.Fatalf("Failed to create temp dir: %v", err)
	}
	defer os.RemoveAll(tempDir)

	// 创建处理单元管理器
	manager := unit.NewProcessingUnitManager()

	// 创建处理单元配置
	config := unit.DefaultProcessingUnitConfig()
	config.ID = "example-unit"
	config.Name = "Example Processing Unit"
	config.StoragePath = tempDir
	config.BatchSize = 10
	config.BatchTimeout = time.Second

	// 创建处理单元
	processingUnit, err := manager.CreateUnit(config)
	if err != nil {
		log.Fatalf("Failed to create processing unit: %v", err)
	}

	// 创建处理器
	// 1. 大写转换处理器
	uppercaseProcessor := processor.NewProcessor("uppercase", processor.MapProcessor(func(record *data.DataRecord) (*data.DataRecord, error) {
		if record.Format == data.DataFormatText {
			record.Value = []byte(strings.ToUpper(string(record.Value)))
		}
		return record, nil
	}), nil)

	// 2. 过滤处理器
	filterProcessor := processor.NewProcessor("filter", processor.FilterProcessor(func(record *data.DataRecord) bool {
		// 过滤掉包含"FILTER"的记录
		return !strings.Contains(string(record.Value), "FILTER")
	}), nil)

	// 3. 添加时间戳处理器
	timestampProcessor := processor.NewProcessor("timestamp", processor.MapProcessor(func(record *data.DataRecord) (*data.DataRecord, error) {
		record.SetHeader("processed_at", time.Now().Format(time.RFC3339))
		return record, nil
	}), nil)

	// 创建处理器链
	chain := processor.NewProcessorChain(uppercaseProcessor, filterProcessor, timestampProcessor)

	// 设置处理器链
	processingUnit.SetProcessorChain(chain)

	// 启动处理单元
	err = processingUnit.Start()
	if err != nil {
		log.Fatalf("Failed to start processing unit: %v", err)
	}
	defer processingUnit.Stop()

	// 创建上下文
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 创建数据生产者
	go produceData(ctx, processingUnit)

	// 创建数据消费者
	go consumeData(ctx, processingUnit)

	// 等待信号
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	log.Println("Processing example running. Press Ctrl+C to stop.")
	<-sigCh
	log.Println("Shutting down...")
}

// produceData 生产数据
func produceData(ctx context.Context, unit *unit.ProcessingUnit) {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	counter := 0
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			// 创建记录
			var key, value string
			if counter%5 == 0 {
				key = fmt.Sprintf("filter-key-%d", counter)
				value = fmt.Sprintf("This is a FILTER test message %d", counter)
			} else {
				key = fmt.Sprintf("key-%d", counter)
				value = fmt.Sprintf("This is test message %d", counter)
			}

			record := data.NewDataRecord([]byte(key), []byte(value), data.DataFormatText)
			record.SetMetadata("source", "producer")
			record.SetMetadata("counter", counter)

			// 发送记录
			err := unit.Send(ctx, record)
			if err != nil {
				log.Printf("Failed to send record: %v", err)
			} else {
				log.Printf("Sent record: %s -> %s", key, value)
			}

			counter++
		}
	}
}

// consumeData 消费数据
func consumeData(ctx context.Context, unit *unit.ProcessingUnit) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			// 接收记录
			record, err := unit.Receive(ctx)
			if err != nil {
				if ctx.Err() != nil {
					return
				}
				log.Printf("Failed to receive record: %v", err)
				time.Sleep(time.Second)
				continue
			}

			// 处理记录
			log.Printf("Received record: %s -> %s", string(record.Key), string(record.Value))
			if processedAt, ok := record.GetHeader("processed_at"); ok {
				log.Printf("  Processed at: %s", processedAt)
			}
			if source, ok := record.GetMetadata("source"); ok {
				log.Printf("  Source: %v", source)
			}
			if counter, ok := record.GetMetadata("counter"); ok {
				log.Printf("  Counter: %v", counter)
			}

			// 存储记录
			err = unit.Store(ctx, record)
			if err != nil {
				log.Printf("Failed to store record: %v", err)
			} else {
				log.Printf("Stored record: %s", string(record.Key))
			}
		}
	}
}
